package com.boot.cluster.worker;

import com.boot.cluster.enums.WorkState;
import com.boot.cluster.worker.bean.TaskResponse;
import com.boot.cluster.worker.bean.TaskRunner;
import com.boot.cluster.worker.bean.TaskSplitterInfo;
import com.boot.cluster.worker.bean.TaskWrap;
import com.boot.cluster.worker.runner.Work;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;


public class Actor implements AutoCloseable{
	private static Logger logger = LoggerFactory.getLogger(Actor.class);
	public Cache<String,Object> cache ;
	private Map<String, Future<TaskResponse>> futureMap = new ConcurrentHashMap<>();
	private Map<String, Work> workMap = new ConcurrentHashMap<>();
	private ExecutorService pool;

	public Actor(ExecutorService pool) {
		this.pool=pool;
		this.cache = CacheBuilder.newBuilder().expireAfterAccess(5, TimeUnit.MINUTES).expireAfterWrite(30,TimeUnit.MINUTES).build();
	}

	/**
	 *	key :id#seq
	 * @param runner
	 * @param info
	 * @return
	 */
	private String getId(TaskRunner runner , TaskSplitterInfo info) {
		if((runner==null) ||(info ==null) ) {
			throw new IllegalArgumentException("task param is null");
		}
		return runner.getId().concat("#").concat(info.getSeq());
	}

	private String getId(TaskResponse resp) {
		if(resp==null) {
			throw new IllegalArgumentException("task resp is null");
		}
		return resp.getId().concat("#").concat(resp.getSeq());
	}

	private  TaskResponse  submit(TaskSplitterInfo info,Work work,TaskResponse resp) {
		String key =this.getId(resp);
		this.futureMap.put(key,this.pool.submit(new ActorThread(work,info,resp)));
		try {
			resp = this.futureMap.get(key).get();
		} catch (InterruptedException | ExecutionException e) {
			resp.setE(new RuntimeException(e));
		}
		this.futureMap.remove(key);
		return resp;
	}


	public List<TaskResponse> submit(List<TaskSplitterInfo> infos,TaskRunner runner) {
		List<TaskResponse> list = new ArrayList<>();
		try {
			Work work = runner.getWcls().getDeclaredConstructor().newInstance(null);
			infos.forEach((info) -> {
				TaskResponse resp = new TaskResponse(runner.getId(), info.getSeq(), runner.getParam());
				list.add(this.submit(info, work, resp));
			});
		} catch (Exception e) {
			logger.error("instantiation error ", e);
		}
		return list;
	}

	public  List<TaskResponse> getResult(TaskWrap wrapper) {
		List<TaskResponse> resp =new ArrayList<>();
		wrapper.getTasks().forEach((s)->{
			TaskResponse rp = (TaskResponse)this.cache.getIfPresent(this.getId(wrapper.getRunner(),s));
			if(rp!=null) {
				resp.add(rp);
			}
		});
		return resp;
	}

	public void cancelAllTask(List<TaskSplitterInfo> infos,TaskRunner runner) {
		infos.forEach((k)->{
			this.cancelTask(runner,k);
		});
	}

	public void cancelTask(TaskRunner runner,TaskSplitterInfo info) {
		String key =this.getId(runner, info);
		if(this.futureMap.containsKey(key)) {
			return;
		}
		boolean isCancel = this.futureMap.get(key).cancel(true);
		logger.info("cancel task key {}, res: {}" ,key, isCancel);
		this.clear(key);
	}

	public void cancelAllTask(TaskRunner runner,List<TaskSplitterInfo> infos) {
		infos.forEach((s)->{
			this.cancelTask(runner,s);
		});
	}

	public void clear(String id ) {
		this.futureMap.remove(id);
		this.workMap.remove(id);
		//		this.cache.invalidate(id);
	}

	public void clean() {
		this.futureMap = null;
		this.workMap=null;
		this.cache.cleanUp();
	}

	@Override
	public void close() throws Exception {
		if(this.pool != null) {
			this.pool.shutdown();
		}
		this.futureMap = null;
		this.workMap=null;
	}

	public	class  ActorThread implements Callable<TaskResponse>{
		private TaskSplitterInfo info;
		private TaskResponse resp;
		private Work work;

		public ActorThread(Work work,TaskSplitterInfo info,TaskResponse resp) {
			this.resp=resp;
			this.info=info;
			this.work=work;
		}

		@Override
		public TaskResponse call() {
			this.resp.setState(WorkState.WORKING);
			try {
				this.resp.setData(this.work.action(this.resp.getParam(),this.info));
			} catch (RuntimeException e) {
				this.resp.setE(e);
				logger.error("action error",e);
			}
			this.resp.setState(WorkState.COMPLETED);
			//保存计算结果
			Actor.this.cache.put(Actor.this.getId(this.resp),this.resp);
			return this.resp;
		}
	}


}
